unit Common.ThreadPool;

interface

uses System.SysUtils, System.Classes, Common.List, System.SyncObjs;

Type
    TTaskItem = class(TObject)
    private
        fTaskName: String;
        fProc: TProc;
    protected
        constructor Create();
    public
        class function Obtain(TaskName: String; Proc: TProc): TTaskItem;
        property TaskName: String read fTaskName write fTaskName;
        property Proc: TProc read fProc write fProc;
    end;

    TThreadPool = class;

    TPoolThread = class(TThread)
    private
        fThreadPool: TThreadPool;
        FEvent: TEvent;
    protected
        procedure Execute(); override;
    public
        constructor Create(ThreadPool: TThreadPool);
        destructor Destroy; override;
        property Event: TEvent read FEvent;
    end;

    TThreadPool = class(TObject)
    private
        ThreadList: TList;
        fTaskList: TLinkedList;
        fCoreSize: Integer;
        fMaxTaskSize: Integer;
    protected
        procedure createThreads;
        procedure DestroyThreads;
    public
        constructor Create(CoreSize: Integer = 1;
          MaxTaskSize: Integer = Integer.MaxValue);
        destructor Destroy; override;
        class function getDefault: TThreadPool;
        procedure Execute(taskItem: TTaskItem); overload;
        procedure Execute(TaskName: String; Proc: TProc); overload;
        procedure Shutdown;
        property CoreSize: Integer read fCoreSize;
        property MaxTaskSize: Integer read fMaxTaskSize;
    end;

implementation

{ TPoolThread }

var
    taskPool: TLinkedList;
    defaultThreadPool: TThreadPool;

constructor TPoolThread.Create(ThreadPool: TThreadPool);
begin
    inherited Create(true);
    FEvent := TEvent.Create();
    fThreadPool := ThreadPool;
end;

destructor TPoolThread.Destroy;
begin
    FEvent.Free;
    inherited;
end;

procedure TPoolThread.Execute;
var
    task: TTaskItem;
begin
    while (Not Terminated) do
    begin
        task := fThreadPool.fTaskList.RemoveFirst;
        if assigned(task) then
        begin
            NameThreadForDebugging('tp-' + task.TaskName);
            if (assigned(task.Proc)) then
            begin
                task.Proc();
                task.fProc:=nil;
            end;
            taskPool.Append(task);
        end
        else
        begin
            if (FEvent.WaitFor() = wrSignaled) then
            begin
                FEvent.ResetEvent;
                if (Terminated) then
                begin
                    break;
                end;
                continue;
            end;
        end;
    end;
end;

{ TThreadPool }

constructor TThreadPool.Create(CoreSize: Integer = 1;
  MaxTaskSize: Integer = Integer.MaxValue);
begin
    inherited Create;
    fTaskList := TLinkedList.Create;
    fCoreSize := CoreSize;
    fMaxTaskSize := MaxTaskSize;
    ThreadList := TList.Create;
    createThreads;
end;

procedure TThreadPool.createThreads;
var
    i: Integer;
    thread: TPoolThread;
begin
    for i := 0 to fCoreSize - 1 do
    begin
        thread := TPoolThread.Create(self);
        ThreadList.Add(thread);
        thread.Start;
    end;
end;

destructor TThreadPool.Destroy;
begin
    Shutdown();
    DestroyThreads;
    ThreadList.Free;
    fTaskList.Free;
    inherited;
end;

procedure TThreadPool.DestroyThreads;
var
    i: Integer;
begin
    for i := 0 to ThreadList.Count - 1 do
    begin
        TPoolThread(ThreadList.Items[i]).Free;
    end;
end;

procedure TThreadPool.Execute(TaskName: String; Proc: TProc);
var
    taskItem: TTaskItem;
begin
    taskItem := TTaskItem.Obtain(TaskName, Proc);
    Execute(taskItem);
end;

class function TThreadPool.getDefault: TThreadPool;
begin
    result := defaultThreadPool;
end;

procedure TThreadPool.Execute(taskItem: TTaskItem);
var
    i, nCount: Integer;
begin
    if fTaskList.Count > MaxTaskSize then
    begin
        raise Exception.Create('Over max task size!')
    end;
    fTaskList.Append(taskItem);
    nCount := ThreadList.Count;
    if fTaskList.Count < ThreadList.Count then
    begin
        nCount := fTaskList.Count mod ThreadList.Count;
    end;
    for i := 0 to nCount - 1 do
    begin
        TPoolThread(ThreadList.Items[i]).Event.SetEvent;
    end;
end;

procedure TThreadPool.Shutdown;
var
    i: Integer;
begin
    for i := 0 to ThreadList.Count - 1 do
    begin
        TPoolThread(ThreadList.Items[i]).Terminate;
        TPoolThread(ThreadList.Items[i]).Event.SetEvent;
    end;
end;

{ TTaskItem }

constructor TTaskItem.Create;
begin
    inherited Create;
end;

class function TTaskItem.Obtain(TaskName: String; Proc: TProc): TTaskItem;
var
    item: TTaskItem;
begin
    item := taskPool.RemoveFirst;
    if not assigned(item) then
    begin
        item := TTaskItem.Create;
    end;
    item.TaskName := TaskName;
    item.Proc := Proc;
    result := item;
end;

procedure clearTaskPool();
var
    taskItem: TTaskItem;
begin
    while not taskPool.IsEmpty do
    begin
        taskItem := TTaskItem(taskPool.RemoveFirst);
        taskItem.Free;
    end;
end;

initialization

taskPool := TLinkedList.Create;

defaultThreadPool := TThreadPool.Create(1);

finalization

defaultThreadPool.Free;
clearTaskPool;
taskPool.Free;

end.
